package org.jeecg.modules.system.util;

import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.Message;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.support.LambdaMeta;
import kotlin.jvm.internal.Lambda;
import org.jeecg.modules.system.entity.EventLog;
import org.jeecg.modules.system.service.IEventLogService;
import org.jeecg.modules.system.service.ISysUserService;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.text.SimpleDateFormat;
import java.util.*;

@Component
public class CanalClient implements InitializingBean {
//public class CanalClient {

    private final static int BATCH_SIZE = 1000;
    private final static String DELETE = "DELETE";
    private final static String INSERT = "INSERT";
    private final static String UPDATE = "UPDATE";

    @Autowired
    private IEventLogService eventLogService;

    @Autowired
    private ISysUserService sysUserService;

    @Value("${jeecg.canal.ip}")
    private String ip;

    @Value("${jeecg.canal.port}")
    private Integer port;

    @Value("${jeecg.canal.username}")
    private String username;

    @Value("${jeecg.canal.password}")
    private String password;

    @Value("${jeecg.canal.destination}")
    private String destination;

    @Value("${jeecg.canal.subscribe}")
    private String subscribe;

    @Override
    public void afterPropertiesSet() throws Exception {
        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(ip,port),destination,username,password);
                try{
                    connector.connect();
                    connector.subscribe(subscribe);
                    connector.rollback();
                    while(true){
                        Message message = connector.getWithoutAck(BATCH_SIZE);
                        long batchId = message.getId();
                        int size = message.getEntries().size();
                        if(batchId == -1 || size ==0){
                            try {
                                Thread.sleep(2000);
                            }catch (Exception e){
                                throw new RuntimeException("系统异常:"+e);
                            }
                        }
                        else{
                            printEntry(message.getEntries());
                        }
                        connector.ack(batchId);
                    }
                }catch (Exception e){
                    throw new RuntimeException("系统异常:"+e);
                }finally {
                    connector.disconnect();
                }
            }
        });
        thread.start();
    }
    /**
     //     * 打印canal server解析binlog获得的实体类信息
     //     */
    private void printEntry(List<Entry> entrys) {
        for (Entry entry : entrys) {
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                //开启/关闭事务的实体类型，跳过
                continue;
            }
            //RowChange对象，包含了一行数据变化的所有特征
            //比如isDdl 是否是ddl变更操作 sql 具体的ddl sql beforeColumns afterColumns 变更前后的数据字段等等
            RowChange rowChage;
            try {
                rowChage = RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
            }
            //获取操作类型：insert/update/delete类型
            EventType eventType = rowChage.getEventType();
            //打印Header信息
            System.out.println(String.format("================》; binlog[%s:%s] , name[%s,%s] , eventType : %s",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                    eventType));
            String schemaName = entry.getHeader().getSchemaName();
            String tableName = entry.getHeader().getTableName();
//            System.out.println("用户登录信息："+loginUser.getRealname());
            //判断是否是DDL语句
            if (rowChage.getIsDdl()) {
                System.out.println("================》;isDdl: true,sql:" + rowChage.getSql());
            }
//            System.out.println("登录的用户："+user.getRealname());
            //获取RowChange对象里的每一行数据，打印出来
            for (RowData rowData : rowChage.getRowDatasList()) {
                if (eventType == EventType.INSERT) {
                    addOrDelete(rowData.getAfterColumnsList(),tableName,INSERT);
                    //如果是更新的语句
                } else {
                    //变更前的数据
//                    System.out.println("------->; before");
//                    printColumn(rowData.getBeforeColumnsList(),tableName,update);
//                    //变更后的数据
//                    System.out.println("------->; after");
//                    printColumn(rowData.getAfterColumnsList(),tableName,update);
                    update(rowData.getBeforeColumnsList(),rowData.getAfterColumnsList(),tableName,schemaName);
                }
            }
        }
    }

    private void addOrDelete(List<Column> columns,String tableName,String flag) {
        String event=null;
        EventLog eventLog = new EventLog();
        eventLog.setTableName(tableName);
        eventLog.setEventType(flag);
        eventLog.setEventTime(new Date());
//        String code = null;
        for (Column column : columns) {
            if(column.getName().equals("update_by") && flag.equals("DELETE")){
                eventLog.setUserName(column.getValue());
                eventLog.setUserId(sysUserService.getIdByName(column.getValue()).getId());
            }else if(column.getName().equals("create_by") && flag.equals("INSERT")){
                eventLog.setUserName(column.getValue());
                eventLog.setUserId(sysUserService.getIdByName(column.getValue()).getId());
            }
//            if(column.getName().equals("code")){
//                code = column.getValue();
//            }
            if(column.getName().equals("id")){
                eventLog.setDataId(column.getValue());
            }
//            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
        }
        Date eventTime = eventLog.getEventTime();
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy年MM月dd日 HH:mm:ss");
        if(flag.equals("INSERT")){
            event = String.format("%s在%s新增了一条ID为%s的数据", eventLog.getUserName(), sdf.format(eventTime), eventLog.getDataId());
        }else
        {
            event = String.format("%s在%s删除了一条ID为%s的数据", eventLog.getUserName(), sdf.format(eventTime), eventLog.getDataId());
        }
        eventLog.setEvent(event);
        eventLogService.save(eventLog);
    }

//    private void delete(List<Column> columns,String tableName) {
//        EventLog eventLog = new EventLog();
//        eventLog.setTableName(tableName);
//        eventLog.setEventTime(new Date());
//        eventLog.setEventType(DELETE);
//        String code = null;
//        for (Column column : columns) {
////            if(column.getName().equals("update_by")){
////                eventLog.setUserName(column.getValue());
////                eventLog.setUserId(sysUserService.getIdByName(column.getValue()).getId());
////            }
//            if(column.getName().equals("code")){
//                code = column.getValue();
//            }
//            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
//        }
//        String event = "删除了一条编码为"+code+"的数据";
//        eventLog.setEvent(event);
//        eventLogService.save(eventLog);
//    }

    private void update(List<Column> beforeColumns,List<Column> afterColumns,String tableName,String schemaName) {
        EventLog eventLog = new EventLog();
        boolean flag = true;
        String filed = null;
        eventLog.setTableName(tableName);
        eventLog.setEventTime(new Date());
        eventLog.setEventType(UPDATE);
//        String code = null;
        List<UpdateValue> list = new ArrayList<>();
//        List<Integer> number = null;
        for(int i=0; i<afterColumns.size(); i++)
        {
            if(afterColumns.get(i).getName().equals("del_flag") && afterColumns.get(i).getUpdated()==true){
                addOrDelete(afterColumns,tableName,DELETE);
                flag = false;
                break;
            }
            if(!afterColumns.get(i).getName().equals("update_time")
                    && !afterColumns.get(i).getName().equals("update_by")
                    && afterColumns.get(i).getUpdated()==true){
                UpdateValue updateValue = new UpdateValue();
                filed = getField(tableName,schemaName,beforeColumns.get(i).getName());
//                System.out.println("filed:中文意思:"+filed);
                updateValue.setFiled(filed);
                updateValue.setBeforeValue(beforeColumns.get(i).getValue());
                updateValue.setAfterValue(afterColumns.get(i).getValue());
                list.add(updateValue);
            }
            if(afterColumns.get(i).getName().equals("update_by")){
                eventLog.setUserName(afterColumns.get(i).getValue());
                eventLog.setUserId(sysUserService.getIdByName(afterColumns.get(i).getValue()).getId());
            }
//            if(afterColumns.get(i).getName().equals("code")){
//                code = afterColumns.get(i).getValue();
//            }
            if(afterColumns.get(i).getName().equals("id")){
                eventLog.setDataId(afterColumns.get(i).getValue());
            }

        }
//        for (Column column : afterColumns) {
//            if(column.getName().equals("code")){
//                code = column.getValue();
//            }
//            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
//        }
        Date eventTime = eventLog.getEventTime();
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy年MM月dd日 HH:mm:ss");
        String event = String.format("%s在%s修改了一条ID为%s的数据", eventLog.getUserName(), sdf.format(eventTime), eventLog.getDataId());
//        String event = "修改了一条编码为"+code+"的数据"+"\n";
        for (UpdateValue updateValue : list) {
            event +=updateValue;
        }
        eventLog.setEvent(event);
        if(flag){
            eventLogService.save(eventLog);
        }
    }

    /**
     *
     * @param tableName  表名
     * @param schemaName  数据库名
     * @param name  字段名
     * @return
     */
    private String getField(String tableName, String schemaName, String name) {
        return eventLogService.getFiled(tableName,schemaName,name);
    }
}
